﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Microsoft.SqlServer.Dts.Pipeline;
using Microsoft.SqlServer.Dts.Pipeline.Wrapper;
using Microsoft.SqlServer.Dts.Runtime.Wrapper;
using System.Xml;
using System.IO;
using System.Security.Policy;
using System.Runtime.Remoting;
using System.Runtime.Remoting.Lifetime;

namespace CompressedFileSource
{
    [DtsPipelineComponent
        (
        DisplayName = "CompressedFileSource", Description = "Inserts rows from a file created by a CompressedFileDest",
        ComponentType = ComponentType.SourceAdapter
        )
        ]
    public class CompressedFileSource : PipelineComponent
    {


        public override void PreExecute()
        {
            base.PreExecute();
            /*
             * Add your code here
             */

            tableName = @"TableData.dat";
            // this is the correct place to open the data file I think:
            // Read the Metadata about the columns in the file
            dataFile = System.IO.Packaging.Package.Open(Connection.ConnectionString.ToString(),
                System.IO.FileMode.Open);



            Uri partUriDocument = System.IO.Packaging.PackUriHelper.CreatePartUri(
                new Uri(@"/" + tableName, UriKind.Relative));
            System.IO.Packaging.PackagePart packagePartDocument =
                dataFile.GetPart(partUriDocument);

            if (!dataFile.PartExists(partUriDocument))
            {
                throw new Exception("The datafile is not valid");
            }
            readStream = new BinaryReader(packagePartDocument.GetStream());


            columnGetters = new DataGetter[ComponentMetaData.OutputCollection[0].OutputColumnCollection.Count];
            Connection = ComponentMetaData.RuntimeConnectionCollection[0].ConnectionManager;
            int i = 0;
            foreach (IDTSOutputColumn100 column in ComponentMetaData.OutputCollection[0].OutputColumnCollection)
            {
                if (column.DataType == DataType.DT_BOOL)
                    columnGetters[i] = new BooleanGetter();
                else if (column.DataType == DataType.DT_BYTES)
                    columnGetters[i] = new BytesGetter();
                else if (column.DataType == DataType.DT_CY || column.DataType == DataType.DT_DECIMAL)
                    columnGetters[i] = new DecimalGetter();
                else if (column.DataType == DataType.DT_DATE)
                    columnGetters[i] = new DateTimeGetter();
                else if (column.DataType == DataType.DT_I8)
                    columnGetters[i] = new Int64Getter();
                else if (column.DataType == DataType.DT_BYTES)
                    columnGetters[i] = new BytesGetter();
                else if (column.DataType == DataType.DT_DBDATE)
                    columnGetters[i] = new DateGetter();
                else if (column.DataType == DataType.DT_DBTIME)
                    columnGetters[i] = new TimeGetter();
                else if (column.DataType == DataType.DT_DBTIME2)
                    columnGetters[i] = new TimeGetter();
                else if (column.DataType == DataType.DT_DBTIMESTAMP || column.DataType == DataType.DT_DBTIMESTAMP2)
                    columnGetters[i] = new DateTimeGetter();
                else if (column.DataType == DataType.DT_DBTIMESTAMPOFFSET)
                    columnGetters[i] = new DateTimeOffsetGetter();
                else if (column.DataType == DataType.DT_FILETIME)
                    columnGetters[i] = new DateTimeGetter();
                else if (column.DataType == DataType.DT_GUID)
                    columnGetters[i] = new GuidGetter();
                else if (column.DataType == DataType.DT_I1 || column.DataType == DataType.DT_UI1)
                    columnGetters[i] = new ByteGetter();
                else if (column.DataType == DataType.DT_I2)
                    columnGetters[i] = new Int16Getter();
                else if (column.DataType == DataType.DT_I4)
                    columnGetters[i] = new Int32Getter();
                else if (column.DataType == DataType.DT_I8)
                    columnGetters[i] = new Int64Getter();
                else if (column.DataType == DataType.DT_NUMERIC)
                    columnGetters[i] = new DecimalGetter();
                else if (column.DataType == DataType.DT_R4)
                    columnGetters[i] = new SingleGetter();
                else if (column.DataType == DataType.DT_R8)
                    columnGetters[i] = new DoubleGetter();
                else if (column.DataType == DataType.DT_BYTES)
                    columnGetters[i] = new BytesGetter();
                else if (column.DataType == DataType.DT_BYTES)
                    columnGetters[i] = new SingleGetter();
                else if (column.DataType == DataType.DT_STR || column.DataType == DataType.DT_WSTR)
                    columnGetters[i] = new StringGetter();
                else if (column.DataType == DataType.DT_UI2)
                    columnGetters[i] = new UInt16Getter();
                else if (column.DataType == DataType.DT_UI4)
                    columnGetters[i] = new UInt32Getter();
                else if (column.DataType == DataType.DT_UI8)
                    columnGetters[i] = new UInt64Getter();
                else if (column.DataType == DataType.DT_IMAGE)
                    columnGetters[i] = new ImageGetter();
                else if (column.DataType == DataType.DT_TEXT || column.DataType == DataType.DT_NTEXT)
                    columnGetters[i] = new TextGetter();
                else columnGetters[i] = new BytesGetter();



                // instantiate the right kind:
                columnGetters[i].colIndex = i;
                columnGetters[i].colLength = column.Length;
                columnGetters[i].colName = column.Name;
                columnGetters[i].precision = column.Precision;
                columnGetters[i].scale = column.Scale;
                columnGetters[i].parent = this;

                // set the reader object:

                columnGetters[i].binReader = readStream;
                // set the columnInformation:

                this.columnInformation = new System.Collections.ArrayList();
                IDTSOutput100 output = ComponentMetaData.OutputCollection[0];
                // set the ColumnInfos in the columnGetters:

                ColumnInfo ci = new ColumnInfo();
                ci.BufferColumnIndex = BufferManager.FindColumnByLineageID(output.Buffer, column.LineageID);
                ci.ColumnName = column.Name;
                columnInformation.Add(ci);

                columnGetters[i].colInfo = ci;
                i++;
            }

        }

        public override void PostExecute()
        {
            base.PostExecute();

            // need to lookup the table name here:
            //ComponentMetaData.FireInformation(10, ComponentMetaData.Name, "Closing file " + this.Connections.Connection.ConnectionString.ToString() + ((curFile == 1) ? "" : (curFile - 1).ToString()), "", 0, false);
            if (readStream != null)
                readStream.Close();
            if (dataFile != null)
                dataFile.Close();
        }

        // this is the method to process the output:
        public override void PrimeOutput(int outputs, int[] outputIDs, PipelineBuffer[] buffers)
        {
            IDTSOutput100 output = ComponentMetaData.OutputCollection[0];
            PipelineBuffer buffer = buffers[0];

            try
            {
                while (readStream.BaseStream.Position < readStream.BaseStream.Length)
                {
                    buffer.AddRow();
                    for (int i = 0; i < columnGetters.Length; i++)
                    {
                        columnGetters[i].ProcessData(buffer);
                    }

                }
            }
            catch (System.IO.IOException e)
            {
                // close the streams and files
                ReleaseConnections();
            }

            buffer.SetEndOfRowset();
        }


        public override void ReinitializeMetaData()
        {
            base.ReinitializeMetaData();

            ComponentMetaData.OutputCollection[0].ExternalMetadataColumnCollection.RemoveAll();
            ComponentMetaData.OutputCollection[0].OutputColumnCollection.RemoveAll();

            // now create the columns anew:
            CreateColumnsFromDataTable();

        }

        public Tuple<String, int, DataType, int, int, int, int>[] GetColumnMetadataFromFile()
        {
            // Read the Metadata about the columns in the file
            bool dataFileIsClosed = (dataFile == null);
            if (dataFileIsClosed)
            {
                dataFile = System.IO.Packaging.Package.Open(Connection.ConnectionString.ToString(),
                    System.IO.FileMode.Open);
            }

            Uri partUriDocument = System.IO.Packaging.PackUriHelper.CreatePartUri(
                new Uri(@"/metaData.xml", UriKind.Relative));
            System.IO.Packaging.PackagePart packagePartDocument =
                dataFile.GetPart(partUriDocument);
            using (System.IO.Stream xmlStream = packagePartDocument.GetStream())
            {
                dom = new XmlDocument();
                XmlReader r = XmlReader.Create(xmlStream);
                dom.Load(r);
                r.Close();
                xmlStream.Close();
            }
            Tuple<String, int, DataType, int, int, int, int>[] columnInfos;
            if (dom == null)
            {
                throw new Exception("The specified file is not a valid CompressedFile created by a CompressedFileDest component");
            }
            else
            {
                /*
                XmlNode xmlnode = dom.CreateNode(XmlNodeType.XmlDeclaration, "", "");
                dom.AppendChild(xmlnode);
                 * */
                // next find the root element:
                XmlNodeList columns = dom.GetElementsByTagName("Column");
                columnInfos = new Tuple<String, int, DataType, int, int, int, int>[columns.Count];
                for (int i = 0; i < columns.Count; i++)
                {
                    Tuple<String, int, DataType, int, int, int, int> columnInfo = new Tuple<String, int, DataType, int, int, int, int>(
                        // need to add info on whether or not it is a BLOB, and handle accordingly.
                    columns.Item(i).Attributes["Name"].InnerText,
                    Convert.ToInt32(columns.Item(i).Attributes["CodePage"].InnerText),
                    getDataType(columns.Item(i).Attributes["DataType"].InnerText),
                    Convert.ToInt32(columns.Item(i).Attributes["Length"].InnerText),
                    Convert.ToInt32(columns.Item(i).Attributes["Precision"].InnerText),
                    Convert.ToInt32(columns.Item(i).Attributes["Scale"].InnerText),
                    Convert.ToInt32(columns.Item(i).Attributes["SortKeyPosition"].InnerText)
                    );
                    columnInfos[i] = columnInfo;
                }
            }
            if (dataFileIsClosed)
                dataFile.Close();
            return columnInfos;

        }



        // function for the DataType function.
        private DataType getDataType(String type_name)
        {
            if (type_name == DataType.DT_BOOL.ToString())
                return DataType.DT_BOOL;
            if (type_name == DataType.DT_BYTES.ToString())
                return DataType.DT_BYTES;
            if (type_name == DataType.DT_DECIMAL.ToString())
                return DataType.DT_DECIMAL;
            if (type_name == DataType.DT_DATE.ToString())
                return DataType.DT_DATE;
            if (type_name == DataType.DT_I8.ToString())
                return DataType.DT_I8;
            if (type_name == DataType.DT_BYTES.ToString())
                return DataType.DT_BYTES;
            if (type_name == DataType.DT_DBDATE.ToString())
                return DataType.DT_DBDATE;
            if (type_name == DataType.DT_DBTIME.ToString())
                return DataType.DT_DBTIME;
            if (type_name == DataType.DT_DBTIME2.ToString())
                return DataType.DT_DBTIME2;
            if (type_name == DataType.DT_DBTIMESTAMP.ToString())
                return DataType.DT_DBTIMESTAMP;
            if (type_name == DataType.DT_DBTIMESTAMP2.ToString())
                return DataType.DT_DBTIMESTAMP2;
            if (type_name == DataType.DT_DBTIMESTAMPOFFSET.ToString())
                return DataType.DT_DBTIMESTAMPOFFSET;
            if (type_name == DataType.DT_FILETIME.ToString())
                return DataType.DT_FILETIME;
            if (type_name == DataType.DT_GUID.ToString())
                return DataType.DT_GUID;
            if (type_name == DataType.DT_I1.ToString())
                return DataType.DT_I1;
            if (type_name == DataType.DT_UI1.ToString())
                return DataType.DT_UI1;
            if (type_name == DataType.DT_I2.ToString())
                return DataType.DT_I2;
            if (type_name == DataType.DT_I4.ToString())
                return DataType.DT_I4;
            if (type_name == DataType.DT_I8.ToString())
                return DataType.DT_I8;
            if (type_name == DataType.DT_NUMERIC.ToString())
                return DataType.DT_NUMERIC;
            if (type_name == DataType.DT_R4.ToString())
                return DataType.DT_R4;
            if (type_name == DataType.DT_R8.ToString())
                return DataType.DT_R8;
            if (type_name == DataType.DT_BYTES.ToString())
                return DataType.DT_BYTES;
            if (type_name == DataType.DT_BYTES.ToString())
                return DataType.DT_BYTES;
            if (type_name == DataType.DT_STR.ToString())
                return DataType.DT_STR;
            if (type_name == DataType.DT_WSTR.ToString())
                return DataType.DT_WSTR;
            if (type_name == DataType.DT_UI2.ToString())
                return DataType.DT_UI2;
            if (type_name == DataType.DT_UI4.ToString())
                return DataType.DT_UI4;
            if (type_name == DataType.DT_UI8.ToString())
                return DataType.DT_UI8;
            if (type_name == DataType.DT_IMAGE.ToString())
                return DataType.DT_IMAGE;
            if (type_name == DataType.DT_TEXT.ToString())
                return DataType.DT_TEXT;
            if (type_name == DataType.DT_NTEXT.ToString())
                return DataType.DT_NTEXT;
            if (type_name == DataType.DT_CY.ToString())
                return DataType.DT_CY;

            // if it gets here, it's not recognized
            throw new Exception("Unsupported pipeline datatype " + type_name);
        }

        public override void ProvideComponentProperties()
        {
            base.ProvideComponentProperties();
            // Reset the component.
            base.RemoveAllInputsOutputsAndCustomProperties();
            ComponentMetaData.RuntimeConnectionCollection.RemoveAll();

            IDTSOutput100 output = ComponentMetaData.OutputCollection.New();
            output.Name = "Output";

            IDTSRuntimeConnection100 connection = ComponentMetaData.RuntimeConnectionCollection.New();
            connection.Name = "CompressedFile";
            

        }

        public override void AcquireConnections(object Transaction)
        {
            if (ComponentMetaData.RuntimeConnectionCollection[0].ConnectionManager != null)
            {
                Microsoft.SqlServer.Dts.Runtime.ConnectionManager conn = Microsoft.SqlServer.Dts.Runtime.DtsConvert.GetWrapper(
                    ComponentMetaData.RuntimeConnectionCollection[0].ConnectionManager);
                Connection = conn.InnerObject as ConnectionManagerFile;
                if (Connection == null)
                    throw new Exception("The Connection Manager " + conn.Name + "is not a FILE connection");

                // now make the connection:

                Connection.AcquireConnection(Transaction);

            }

        }

        public override void ReleaseConnections()
        {
            base.ReleaseConnections();
            if (readStream != null)
                readStream.Close();
            if (dataFile != null)
                dataFile.Close();
            if (ComponentMetaData.RuntimeConnectionCollection[0] != null && ComponentMetaData.RuntimeConnectionCollection[0].ConnectionManager != null)
                ComponentMetaData.RuntimeConnectionCollection[0].ConnectionManager.ReleaseConnection(null);
        }

        private void CreateColumnsFromDataTable()
        {
            // Get the output.
            IDTSOutput100 output = ComponentMetaData.OutputCollection[0];

            // Start clean, and remove the columns from both collections.
            output.OutputColumnCollection.RemoveAll();
            output.ExternalMetadataColumnCollection.RemoveAll();

            Tuple<String, int, DataType, int, int, int, int>[] columnInfos = GetColumnMetadataFromFile();

            // Walk the columns in the schema, 
            // and for each data column create an output column and an external metadata column.
            for (int i = 0; i < columnInfos.Length; i++)
            {
                IDTSOutputColumn100 outColumn = output.OutputColumnCollection.New();
                //IDTSExternalMetadataColumn100 exColumn = output.ExternalMetadataColumnCollection.New();

                // Set column data type properties.
                DataType dt = columnInfos[i].Item3;

                int length = columnInfos[i].Item4;
                int precision = columnInfos[i].Item5;
                int scale = columnInfos[i].Item6;
                int codepage = columnInfos[i].Item2;


                // Set the properties of the output column.
                outColumn.Name = columnInfos[i].Item1;
                outColumn.SortKeyPosition = columnInfos[i].Item7;
                outColumn.SetDataTypeProperties(dt, length, precision, scale, codepage);
                CreateExternalMetaDataColumn(output, outColumn);
            }
        }

        // function to create the external metadata column from the output column

        private void CreateExternalMetaDataColumn(IDTSOutput100 output, IDTSOutputColumn100 outputColumn)
        {

            // Set the properties of the external metadata column.
            IDTSExternalMetadataColumn100 externalColumn = output.ExternalMetadataColumnCollection.New();
            externalColumn.Name = outputColumn.Name;
            externalColumn.Precision = outputColumn.Precision;
            externalColumn.Length = outputColumn.Length;
            externalColumn.DataType = outputColumn.DataType;
            externalColumn.Scale = outputColumn.Scale;

            // Map the external column to the output column.
            outputColumn.ExternalMetadataColumnID = externalColumn.ID;

        }


        public override DTSValidationStatus Validate()
        {
            bool pbCancel = false;

            // Validate that there is one output.
            if (ComponentMetaData.OutputCollection.Count != 1)
            {
                ComponentMetaData.FireError(0, ComponentMetaData.Name, "Incorrect number of outputs.", "", 0, out pbCancel);
                return DTSValidationStatus.VS_ISCORRUPT;
            }

            // validate that there is one connection manager:
            if (ComponentMetaData.RuntimeConnectionCollection.Count != 1)
            {
                ComponentMetaData.FireError(0, ComponentMetaData.Name, "Incorrect number of Connection Managers.", "", 0, out pbCancel);
                return DTSValidationStatus.VS_ISBROKEN;

            }


            // Let the base class verify that the input column reflects the output 
            // of the upstream component.

            DTSValidationStatus retval = base.Validate();

            return retval;
        }


        int curFile;
        int rowsInCurrentFile = 1;
        System.IO.Packaging.Package dataFile;
        System.IO.Packaging.PackagePart dataPart;
        String tableName;
        XmlDocument dom;
        DataGetter[] columnGetters;
        IDTSConnectionManager100 Connection;
        System.IO.BinaryReader readStream;
        System.Collections.ArrayList columnInformation;

        private struct ColumnInfo
        {
            public int BufferColumnIndex;
            public string ColumnName;
        }


        private class DataGetter
        {
            public String colName;
            public int colLength;
            public int colIndex;
            public int precision;
            public int scale;
            public System.IO.BinaryReader binReader;
            public System.IO.Packaging.Package dataFile;
            public CompressedFileSource parent;
            public long rowsInCurrentFile = 0;
            public bool isNull;

            public ColumnInfo colInfo;

            public virtual void ProcessData(PipelineBuffer buf)
            {

            }

            public void processNewFile(ref System.IO.Packaging.Package newDataFile, ref System.IO.BinaryReader newBinReader)
            {
                dataFile = newDataFile;
                binReader = newBinReader;
                rowsInCurrentFile = 0;
            }
        }
        private class Int16Getter : CompressedFileSource.DataGetter
        {
            public override void ProcessData(PipelineBuffer buf)
            {

                isNull = binReader.ReadBoolean();
                if (isNull)
                {
                    buf.SetNull(colInfo.BufferColumnIndex);
                }
                else
                {
                    buf.SetInt16(colInfo.BufferColumnIndex, binReader.ReadInt16());
                }

            }
        }

        private class Int32Getter : CompressedFileSource.DataGetter
        {
            public override void ProcessData(PipelineBuffer buf)
            {

                isNull = binReader.ReadBoolean();
                if (isNull)
                {
                    buf.SetNull(colInfo.BufferColumnIndex);
                }
                else
                {
                    buf.SetInt32(colInfo.BufferColumnIndex, binReader.ReadInt32());
                }

            }
        }
        private class Int64Getter : CompressedFileSource.DataGetter
        {
            public override void ProcessData(PipelineBuffer buf)
            {

                isNull = binReader.ReadBoolean();
                if (isNull)
                {
                    buf.SetNull(colInfo.BufferColumnIndex);
                }
                else
                {
                    buf.SetInt64(colInfo.BufferColumnIndex, binReader.ReadInt64());
                }

            }
        }
        private class SByteGetter : CompressedFileSource.DataGetter
        {
            public override void ProcessData(PipelineBuffer buf)
            {

                isNull = binReader.ReadBoolean();
                if (isNull)
                {
                    buf.SetNull(colInfo.BufferColumnIndex);
                }
                else
                {
                    buf.SetSByte(colInfo.BufferColumnIndex, binReader.ReadSByte());
                }

            }
        }
        private class SingleGetter : CompressedFileSource.DataGetter
        {
            public override void ProcessData(PipelineBuffer buf)
            {

                isNull = binReader.ReadBoolean();
                if (isNull)
                {
                    buf.SetNull(colInfo.BufferColumnIndex);
                }
                else
                {
                    buf.SetSingle(colInfo.BufferColumnIndex, binReader.ReadSingle());
                }

            }
        }
        private class StringGetter : CompressedFileSource.DataGetter
        {
            public override void ProcessData(PipelineBuffer buf)
            {

                isNull = binReader.ReadBoolean();
                if (isNull)
                {
                    buf.SetNull(colInfo.BufferColumnIndex);
                }
                else
                {
                     buf.SetString(colInfo.BufferColumnIndex, binReader.ReadString());
                }

            }
        }
        private class TimeGetter : CompressedFileSource.DataGetter
        {
            public override void ProcessData(PipelineBuffer buf)
            {

                isNull = binReader.ReadBoolean();
                if (isNull)
                {
                    buf.SetNull(colInfo.BufferColumnIndex);
                }
                else
                {
                    buf.SetTime(colInfo.BufferColumnIndex, new TimeSpan(binReader.ReadInt64()));
                }

            }
        }
        private class UInt16Getter : CompressedFileSource.DataGetter
        {
            public override void ProcessData(PipelineBuffer buf)
            {

                isNull = binReader.ReadBoolean();
                if (isNull)
                {
                    buf.SetNull(colInfo.BufferColumnIndex);
                }
                else
                {
                    buf.SetUInt16(colInfo.BufferColumnIndex, binReader.ReadUInt16());
                }

            }

        }
        private class UInt32Getter : CompressedFileSource.DataGetter
        {
            public override void ProcessData(PipelineBuffer buf)
            {

                isNull = binReader.ReadBoolean();
                if (isNull)
                {
                    buf.SetNull(colInfo.BufferColumnIndex);
                }
                else
                {
                    buf.SetUInt32(colInfo.BufferColumnIndex, binReader.ReadUInt32());
                }

            }
        }
        private class UInt64Getter : CompressedFileSource.DataGetter
        {
            public override void ProcessData(PipelineBuffer buf)
            {

                isNull = binReader.ReadBoolean();
                if (isNull)
                {
                    buf.SetNull(colInfo.BufferColumnIndex);
                }
                else
                {
                    buf.SetUInt64(colInfo.BufferColumnIndex, binReader.ReadUInt64());
                }

            }
        }
        private class GuidGetter : CompressedFileSource.DataGetter
        {
            public override void ProcessData(PipelineBuffer buf)
            {

                isNull = binReader.ReadBoolean();
                if (isNull)
                {
                    buf.SetNull(colInfo.BufferColumnIndex);
                }
                else
                {
                    buf.SetGuid(colInfo.BufferColumnIndex, new Guid(binReader.ReadString()));
                }

            }
        }
        private class DoubleGetter : CompressedFileSource.DataGetter
        {
            public override void ProcessData(PipelineBuffer buf)
            {

                isNull = binReader.ReadBoolean();
                if (isNull)
                {
                    buf.SetNull(colInfo.BufferColumnIndex);
                }
                else
                {
                    buf.SetDouble(colInfo.BufferColumnIndex, binReader.ReadDouble());
                }

            }

        }
        private class DecimalGetter : CompressedFileSource.DataGetter
        {
            public override void ProcessData(PipelineBuffer buf)
            {

                isNull = binReader.ReadBoolean();
                if (isNull)
                {
                    buf.SetNull(colInfo.BufferColumnIndex);
                }
                else
                {
                    buf.SetDecimal(colInfo.BufferColumnIndex, binReader.ReadDecimal());
                }

            }

        }
        // I'm not yet sure if the DateTimeOffset is working right. I might not be getting all the data.
        private class DateTimeOffsetGetter : CompressedFileSource.DataGetter
        {
            public override void ProcessData(PipelineBuffer buf)
            {

                isNull = binReader.ReadBoolean();
                if (isNull)
                {
                    buf.SetNull(colInfo.BufferColumnIndex);
                }
                else
                {
                    buf.SetDateTimeOffset(colInfo.BufferColumnIndex, new DateTimeOffset(new DateTime(binReader.ReadInt64())));
                }

            }
        }
        private class DateTimeGetter : CompressedFileSource.DataGetter
        {
            public override void ProcessData(PipelineBuffer buf)
            {

                isNull = binReader.ReadBoolean();
                if (isNull)
                {
                    buf.SetNull(colInfo.BufferColumnIndex);
                }
                else
                {
                    buf.SetDateTime(colInfo.BufferColumnIndex, new DateTime(binReader.ReadInt64()));
                }

            }
        }
        private class DateGetter : CompressedFileSource.DataGetter
        {
            public override void ProcessData(PipelineBuffer buf)
            {

                isNull = binReader.ReadBoolean();
                if (isNull)
                {
                    buf.SetNull(colInfo.BufferColumnIndex);
                }
                else
                {
                    buf.SetDate(colInfo.BufferColumnIndex, new DateTime(binReader.ReadInt64()));
                }

            }
        }

        private class BytesGetter : CompressedFileSource.DataGetter
        {
            public override void ProcessData(PipelineBuffer buf)
            {

                isNull = binReader.ReadBoolean();
                if (isNull)
                {
                    buf.SetNull(colInfo.BufferColumnIndex);
                }
                else
                {
                    int length = binReader.ReadInt32();
                    buf.SetBytes(colInfo.BufferColumnIndex, binReader.ReadBytes(length));
                }

            }
        }
        private class ByteGetter : CompressedFileSource.DataGetter
        {
            public override void ProcessData(PipelineBuffer buf)
            {

                isNull = binReader.ReadBoolean();
                if (isNull)
                {
                    buf.SetNull(colInfo.BufferColumnIndex);
                }
                else
                {
                    buf.SetByte(colInfo.BufferColumnIndex, binReader.ReadByte());
                }

            }

        }
        private class BooleanGetter : CompressedFileSource.DataGetter
        {
            public override void ProcessData(PipelineBuffer buf)
            {

                isNull = binReader.ReadBoolean();
                if (isNull)
                {
                    buf.SetNull(colInfo.BufferColumnIndex);
                }
                else
                {
                    buf.SetBoolean(colInfo.BufferColumnIndex, binReader.ReadBoolean());
                }

            }
        }
        private class TextGetter : CompressedFileSource.DataGetter
        {
            public override void ProcessData(PipelineBuffer buf)
            {

                isNull = binReader.ReadBoolean();
                if (isNull)
                {
                    buf.SetNull(colInfo.BufferColumnIndex);
                }
                else
                {
                    blobSize = binReader.ReadUInt32();
                    byte[] buffer;
                    bytesInArray = 1024;
                    offset = 0;
                    while (offset < blobSize)
                    {
                        uint nextBufferLength = (uint)((blobSize - offset > bytesInArray) ? bytesInArray : blobSize - offset);
                        buffer = binReader.ReadBytes((int)nextBufferLength);
                        buf.AddBlobData(colInfo.BufferColumnIndex, buffer);
                        offset += buffer.Length;
                    }
                }
            }
            uint blobSize;
            int bytesInArray = 1024;
            int offset = 0;
        }
        private class ImageGetter : CompressedFileSource.DataGetter
        {
            public override void ProcessData(PipelineBuffer buf)
            {

                isNull = binReader.ReadBoolean();
                if (isNull)
                {
                    buf.SetNull(colInfo.BufferColumnIndex);
                }
                else
                {
                    blobSize = binReader.ReadUInt32();
                    byte[] buffer;
                    bytesInArray = 1024;
                    offset = 0;
                    while (offset < blobSize)
                    {
                        uint nextBufferLength = (uint)((blobSize - offset > bytesInArray) ? bytesInArray : blobSize - offset);
                        buffer = binReader.ReadBytes((int)nextBufferLength);
                        buf.AddBlobData(colInfo.BufferColumnIndex, buffer);
                        offset += buffer.Length;
                    }
                }
            }
            uint blobSize;
            int bytesInArray = 1024;
            int offset = 0;
        }
    }

}